package com.itheima.article.config;

import org.apache.kafka.streams.StreamsBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * KafkaStreamListener扫描和实例化成KafkaStreamProcessor.doAction的返回类，完成监听器实际注册的过程
 */
@Component
public class KafkaStreamListenerFactory implements InitializingBean {

    Logger logger = LoggerFactory.getLogger(com.itheima.article.config.KafkaStreamListenerFactory.class);

    //容器本身
    @Autowired
    DefaultListableBeanFactory defaultListableBeanFactory;

    /**
     * 初始化完成后自动调用
     */
    @Override
    public void afterPropertiesSet() {
        //从容器中根据类型来获取接口的实例对象 可能有多个  bean的话是有bean的id  <bean id="xxxx" name="xxx">
        //key: id的值（bean名称）  beanName
        //value: 类型对应的实例对象
        Map<String, com.itheima.article.config.KafkaStreamListener> map = defaultListableBeanFactory.getBeansOfType(com.itheima.article.config.KafkaStreamListener.class);
        for (String key : map.keySet()) {
            com.itheima.article.config.KafkaStreamListener k = map.get(key);
            //
            com.itheima.article.config.KafkaStreamProcessor processor = new com.itheima.article.config.KafkaStreamProcessor(
                    //从容器获取StreamsBuilder对象实例
                    defaultListableBeanFactory.getBean(StreamsBuilder.class)// 参数1
                    ,k //参数2
            );
            //设置一个bean的名称
            String beanName = k.getClass().getSimpleName()+"AutoProcessor" ;
            //执行注册的动作（将对象交给spring容器管理）
            defaultListableBeanFactory.registerSingleton(beanName,processor.doAction());
            logger.info("add kafka stream auto listener [{}]",beanName);
        }
    }

}
